package chapter4.eg12;

import chapter4.AbstractProducer;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.junit.Test;

public class InterceptorTest extends AbstractProducer {

  @Test
  public void testInterceptor() throws ExecutionException, InterruptedException {
    List<String> interceptors = new ArrayList<>();
    interceptors.add("chapter4.eg12.TimestampPrependerInterceptor");
    interceptors.add("chapter4.eg12.CounterInterceptor");
    properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);

    String topic = "test-topic";
    Producer<String, String> producer = new KafkaProducer<>(properties);
    for (int i = 0; i < 10; i++) {
      ProducerRecord<String, String> record = new ProducerRecord<>(topic, "message" + i);
      RecordMetadata metadata = producer.send(record).get();
      printMetadata(metadata);
    }

    producer.close();
  }
}
